/*
 * Copyright (C) 2010 Seapeak.Xu
 *
 * fostmq may be copied only under the terms of the GNU General
 * Public License V3, which may be found in the fostmq source kit.
 *
 * Created on: 2010-8-17
 * Author: Seapeak.Xu
 */

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <event.h>
#include <errno.h>
#include <unistd.h>
#include <sys/socket.h>
#include <string.h>
#include <sys/types.h>

#include "fost_logger.h"
#include "fost_msg.h"
#include "fost_sock.h"
#include "fost_event.h"
#include "fost_taskpool.h"
#include "fost_mem.h"

/*
 * client send the net stream buffer with a request
 * parameters:
 * 	sock:the client sockfd
 *	events:the trigger event type
 *	arg:the callback function parameter,it type the task_t
 */
void client_request_event(int sock, short events, void *arg)
{
	if (NULL == arg)
	{
		log_error("File:%s,Line:%d,Function:%s.the argument is null.",
				__FILE__, __LINE__, __FUNCTION__);
		return;
	}

	task_t *task = (task_t *) arg;
	if (EV_TIMEOUT == events)
	{
		log_error("File:%s,Line:%d,Function:%s.write event is timeout.",
				__FILE__, __LINE__, __FUNCTION__);
		return;
	}

	int result = 0;

	while(header == task->session_type)
	{
		if (-1 == (result = send(task->sock, (task->req_headerbuff + task->offset), (HEADERLEN - task->offset), 0)))
		{
			if (EAGAIN == errno || EWOULDBLOCK == errno)
			{
				if (0 != event_add(&(task->req_handler), &(task->timeout)))
				{
					log_error("File:%s,Line:%d,Function:%s.add write to event base is error.Errno:%d,Info:%s.",
							__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
					return;
				}
			}
			else
			{
				log_error("File:%s,Line:%d,Function:%s.send response info to client is error.Errno:%d,Info:%s.",
						__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
				return;
			}
			return;
		}

		task->offset += result;

		if (HEADERLEN > task->offset)
		{
			if (0 != event_add(&(task->req_handler), &(task->timeout)))
			{
				log_error("File:%s,Line:%d,Function:%s.add write event to event base is error.",
						__FILE__, __LINE__, __FUNCTION__);
				return;
			}
		}

		if (HEADERLEN == task->offset)
		{
			if (0 == task->req_header.bodylen || NULL == task->req_bodybuff || 0 == strlen(task->req_bodybuff))
			{
				return;
			}

			task->session_type = body;
			task->offset = 0;
			break;
		}
	}

	while(body == task->session_type)
	{
		if (0 == task->req_header.bodylen || NULL == task->req_bodybuff || 0 == strlen(task->req_bodybuff))
		{
			return;
		}

		if (-1 == (result = send(task->sock,(task->req_bodybuff + task->offset), (task->req_header.bodylen - task->offset), 0)))
		{
			if (EAGAIN == errno || EWOULDBLOCK == errno)
			{
				if (0 != event_add(&(task->req_handler), &(task->timeout)))
				{
					log_error("File:%s,Line:%d,Function:%s.add write to event base is error.Errno:%d,Info:%s.",
							__FILE__, __LINE__, __FUNCTION__,errno, strerror(errno));
					return;
				}
			}
			else
			{
				log_error("File:%s,Line:%d,Function:%s.send response info to client is error.Errno:%d,Info:%s.",
						__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
				return;
			}
		}

		task->offset += result;
		if (task->req_header.bodylen > task->offset)
		{
			if (0 != event_add(&(task->req_handler), &(task->timeout)))
			{
				log_error("File:%s,Line:%d,Function:%s.add write event to event base is error.",
						__FILE__, __LINE__, __FUNCTION__);
				return;
			}
		}
		if (task->req_header.bodylen == task->offset)
		{
			return;
		}
	}

}


/*
 * client read the response stream buffer
 * parameters:
 *	sock:the client sockfd
 *	events:the trigger event type
 *	arg:the callback function parameter,it type is task_t
 */
void client_response_event(int sock, short events, void *arg)
{
	if (NULL == arg)
	{
		log_error("File:%s,Line:%d,Function:%s.the argument is null.",
				__FILE__, __LINE__, __FUNCTION__);
		return;
	}
	task_t *task = (task_t *) arg;

	if (EV_TIMEOUT == events && 0 != task->sock)//time out
	{
		if (0 != event_add(&(task->resp_handler), &(task->timeout)))
		{
			log_error(
					"File:%s,Line:%d,Function:%s.add timeout event for %d sockfd with read event is error.Errno:%d,Info:%s.",
					__FILE__, __LINE__, __FUNCTION__,sock,errno, strerror(errno));

		}
		return;
	}

	int bytes = 0;

	while (header == task->session_type)//recv header
	{
		bytes = recv(task->sock, task->resp_headerbuff + task->offset,
				HEADERLEN - task->offset, 0);
		if (0 > bytes)
		{
			if (errno == EAGAIN || errno == EWOULDBLOCK)
			{
				if (0 != event_add(&(task->resp_handler), &(task->timeout)))
				{
					log_error("File:%s,Line:%s,Function:%s.add read event to event_base error.Errno:%d,Info:%s.",
							__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
					return;
				}
			}
			else
			{
				log_error(
						"File:%s,Line:%d,Function:%s.recv is failed when read header data.Errno:%d,Info:%s.",
						__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
				return;
			}
			return;
		}
		else if (0 == bytes)
		{
			log_warn(
					"File:%s,Line:%d,Function:%s.it is not recving  anything data from sock when recv header buffer.",
					__FILE__, __LINE__, __FUNCTION__);
			return;
		}

		if (HEADERLEN > bytes + task->offset)
		{
			task->offset += bytes;
			if (0 != event_add(&(task->resp_handler), &(task->timeout)))
			{
				log_error(
						"File:%s,Line:%s,Function:%s.add the read event to event_base again is error.Errno:%d,Info:%s.",
						__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
				return;
			}
		}
		else
		{
			task->session_type = body;
			task->header_proxy(task);

			if(0 != task->resp_header.bodylen)
			{
				task->resp_bodybuff = fost_callocs(task->resp_header.bodylen);
				if (NULL == task->resp_bodybuff)
				{
					log_error(
							"File:%s,Line:%d,Function:%s.malloc momey to task body_buffer is error.Errno:%d,Info:%s.",
							__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
					return;
				}
				task->offset = 0;//set recv body buffer offset to 0
			}
			break;
		}
	}

	while (0 != task->resp_header.bodylen && body == task->session_type)
	{
		bytes = recv(task->sock, task->resp_bodybuff + task->offset,
				task->resp_header.bodylen - task->offset, 0);
		if (0 > bytes)
		{
			if (errno == EAGAIN || errno == EWOULDBLOCK)
			{
				if (0 != event_add(&(task->resp_handler), &(task->timeout)))
				{
					log_error("File:%s,Line:%d,Function:%s.add event to event_base error when read body buffer.Errno:%d,Info:%s.",
							__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
					return;
				}
			}
			else
			{
				log_error(
						"File:%s,Line:%d,Function:%s.recv is failed when read body buffer.Errno:%d,Info:%s.",
						__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
				return;
			}
			return;
		}
		else if (0 == bytes)
		{
			log_warn("File:%s,Line:%d,Function:%s.recv is fail.Errno:%d,Info:%s.",
					__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
			return;
		}

		if (task->resp_header.bodylen - task->offset > bytes)
		{
			task->offset += bytes;
			if (0 != event_add(&(task->resp_handler), &(task->timeout)))
			{
				log_error("File:%s,Line:%d,Function:%s.add the read event is fail when read body buffer.Errno:%d,Info:%s.",
						__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
				return;
			}
		}
		else
		{
			task->session_type = unspecified;
			*(task->resp_bodybuff + task->resp_header.bodylen) = '\0';
			break;
		}
	}

	task->body_proxy(task);
	return;
}

void client_work_init(struct task_elt_t *task)
{

	log_debug("File:%s,Line:%d,Function:%s.task idx is %d..",__FILE__,__LINE__,__FUNCTION__,task->index);
	if (NULL == task)
	{
		log_error("File:%s,Line:%d,Function:%s.the thread argument is null.",
				__FILE__, __LINE__, __FUNCTION__);
		return;
	}

	if (0 >= task->sock)
	{
		log_error(
				"File:%s,Line:%d,Function:%s.the client socket fd is less 0.the task index is %d, sockfd is %d.",
				__FILE__, __LINE__, __FUNCTION__,task->index, task->sock);
		return;
	}

	if (-1 == set_nonblocking(task->sock))
	{
		log_error("File:%s,Line:%d,Function:%s.set the socket is error.",__FILE__,__LINE__,__FUNCTION__);
		return;
	}

	do
	{
		task->session_type = header;
		event_set(&(task->req_handler), task->sock, EV_WRITE, client_request_event,
				(void *) task);

		if (-1 == event_base_set(task->event_base, &(task->req_handler)))
		{
			log_error(
					"File:%s,Line:%d,Function:%s.Associate the read_header event to event_base is error.Errno:%d,Info:%s",
					__FILE__, __LINE__, __FUNCTION__,errno, strerror(errno));
			task->session_type = unspecified;
			break;
		}

		event_set(&(task->resp_handler), task->sock, EV_READ, client_response_event,
				(void *) task);

		if (0 != event_base_set(task->event_base, &(task->resp_handler)))
		{
			log_error(
					"File:%s,Line:%d,Function:%s.Associate the write hander to event_base is error.Errno:%d,Info:%s.",
					__FILE__, __LINE__, __FUNCTION__,errno,strerror(errno));
			task->session_type = unspecified;
			break;
		}

		if (0 != event_add(&(task->req_handler), &(task->timeout)))
		{
			log_error(
					"File:%s,Line:%d,Function:%s.Add the read header event to  event_base is error.Errno:%d,Info:%s.",
					__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
			log_error("sockfd:%d.index:%d.",task->sock,task->index);
			task->session_type = unspecified;
			break;
		}

		event_base_loop(task->event_base, 0);
	} while (false);

	return;
}



//begin the srever function

/*
 * read the net stream buffer
 * parameters:
 *	sock:the client sockfd
 *	events:the trigger event type
 *	arg:the callback function parameter,it type is task_t
 */
void server_request_event(int sock, short events, void *arg)
{
	if (NULL == arg)
	{
		log_error("File:%s,Line:%d,Function:%s.the argument is null.",
				__FILE__, __LINE__, __FUNCTION__);
		return;
	}
	task_t *task = (task_t *) arg;

	if (EV_TIMEOUT == events && 0 != task->sock)//time out
	{
		if (0 != event_add(&(task->req_handler), &(task->timeout)))
		{
			log_error(
					"File:%s,Line:%d,Function:%s.add timeout event for %d sockfd with read event is error.Errno:%d,Info:%s.",
					__FILE__, __LINE__, __FUNCTION__,sock,errno, strerror(errno));

		}
		return;
	}

	int bytes = 0;

	while (header == task->session_type)//recv header
	{
		bytes = recv(task->sock, task->req_headerbuff + task->offset,
				HEADERLEN - task->offset, 0);
		if (0 > bytes)
		{
			if (errno == EAGAIN || errno == EWOULDBLOCK)
			{
				if (0 != event_add(&(task->req_handler), &(task->timeout)))
				{
					log_error("File:%s,Line:%s,Function:%s.add read event to event_base error.Errno:%d,Info:%s.",
							__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
					return;
				}
			}
			else
			{
				log_error(
						"File:%s,Line:%d,Function:%s.recv is failed when read header data.Errno:%d,Info:%s.",
						__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
				return;
			}
			return;
		}
		else if (0 == bytes)
		{
			log_warn(
					"File:%s,Line:%d,Function:%s.it is not recving  anything data from sock when recv header buffer.",
					__FILE__, __LINE__, __FUNCTION__);
			return;
		}

		if (HEADERLEN > bytes + task->offset)
		{
			task->offset += bytes;
			if (0 != event_add(&(task->req_handler), &(task->timeout)))
			{
				log_error(
						"File:%s,Line:%s,Function:%s.add the read event to event_base again is error.Errno:%d,Info:%s.",
						__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
				return;
			}
		}
		else
		{
			task->session_type = body;
			task->header_proxy(task);

			if(0 != task->req_header.bodylen)
			{
				task->req_bodybuff = fost_callocs(task->req_header.bodylen);
				if (NULL == task->req_bodybuff)
				{
					log_error(
							"File:%s,Line:%d,Function:%s.malloc momey to task body_buffer is error.Errno:%d,Info:%s.",
							__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
					return;
				}
				task->offset = 0;//set recv body buffer offset to 0
			}
			break;
		}
	}

	while (0 != task->req_header.bodylen && body == task->session_type)
	{
		bytes = recv(task->sock, task->req_bodybuff + task->offset,
				task->req_header.bodylen - task->offset, 0);
		if (0 > bytes)
		{
			if (errno == EAGAIN || errno == EWOULDBLOCK)
			{
				if (0 != event_add(&(task->req_handler), &(task->timeout)))
				{
					log_error("File:%s,Line:%d,Function:%s.add event to event_base error when read body buffer.Errno:%d,Info:%s.",
							__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
					return;
				}
			}
			else
			{
				log_error(
						"File:%s,Line:%d,Function:%s.recv is failed when read body buffer.Errno:%d,Info:%s.",
						__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
				return;
			}
			return;
		}
		else if (0 == bytes)
		{
			log_warn("File:%s,Line:%d,Function:%s.recv is fail.Errno:%d,Info:%s.",
					__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
			return;
		}

		if (task->req_header.bodylen - task->offset > bytes)
		{
			task->offset += bytes;
			if (0 != event_add(&(task->req_handler), &(task->timeout)))
			{
				log_error("File:%s,Line:%d,Function:%s.add the read event is fail when read body buffer.Errno:%d,Info:%s.",
						__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
				return;
			}
		}
		else
		{
			task->session_type = unspecified;
			*(task->req_bodybuff + task->req_header.bodylen) = '\0';
			break;
		}
	}

	task->body_proxy(task);
	return;
}

/*
 * send the net stream buffer
 * parameters:
 *	sock:the client sockfd
 *	events:the trigger event type
 *	arg:the callback function parameter,it type the task_t
 */
void server_response_event(int sock, short events, void *arg)
{
	if (NULL == arg)
	{
		log_error("File:%s,Line:%d,Function:%s.the argument is null.",
				__FILE__, __LINE__, __FUNCTION__);
		return;
	}

	task_t *task = (task_t *) arg;
	if (EV_TIMEOUT == events)
	{
		log_error("File:%s,Line:%d,Function:%s.write event is timeout.",
				__FILE__, __LINE__, __FUNCTION__);
		return;
	}

	int result = 0;

	while(header == task->session_type)
	{
		if (-1 == (result = send(task->sock, (task->resp_headerbuff + task->offset), (HEADERLEN - task->offset), 0)))
		{
			if (EAGAIN == errno || EWOULDBLOCK == errno)
			{
				if (0 != event_add(&(task->resp_handler), &(task->timeout)))
				{
					log_error("File:%s,Line:%d,Function:%s.add write to event base is error.Errno:%d,Info:%s.",
							__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
					return;
				}
			}
			else
			{
				log_error("File:%s,Line:%d,Function:%s.send response info to client is error.Errno:%d,Info:%s.",
						__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
				return;
			}
			return;
		}

		task->offset += result;

		if (HEADERLEN > task->offset)
		{
			if (0 != event_add(&(task->resp_handler), &(task->timeout)))
			{
				log_error("File:%s,Line:%d,Function:%s.add write event to event base is error.",
						__FILE__, __LINE__, __FUNCTION__);
				return;
			}
		}

		if (HEADERLEN == task->offset)
		{
			if (0 == task->resp_header.bodylen || NULL == task->resp_bodybuff || 0 == strlen(task->resp_bodybuff))
			{
				return;
			}

			task->session_type = body;
			task->offset = 0;
			break;
		}
	}

	while(body == task->session_type)
	{
		if (0 == task->resp_header.bodylen || NULL == task->resp_bodybuff || 0 == strlen(task->resp_bodybuff))
		{
			return;
		}

		if (-1 == (result = send(task->sock,(task->resp_bodybuff + task->offset), (task->resp_header.bodylen - task->offset), 0)))
		{
			if (EAGAIN == errno || EWOULDBLOCK == errno)
			{
				if (0 != event_add(&(task->resp_handler), &(task->timeout)))
				{
					log_error("File:%s,Line:%d,Function:%s.add write to event base is error.Errno:%d,Info:%s.",
							__FILE__, __LINE__, __FUNCTION__,errno, strerror(errno));
					return;
				}
			}
			else
			{
				log_error("File:%s,Line:%d,Function:%s.send response info to client is error.Errno:%d,Info:%s.",
						__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
				return;
			}
		}

		task->offset += result;
		if (task->resp_header.bodylen > task->offset)
		{
			if (0 != event_add(&(task->resp_handler), &(task->timeout)))
			{
				log_error("File:%s,Line:%d,Function:%s.add write event to event base is error.",
						__FILE__, __LINE__, __FUNCTION__);
				return;
			}
		}
		if (task->resp_header.bodylen == task->offset)
		{
			return;
		}
	}

}

/*
 * begin the thread proxy function
 */
void server_work_init(struct task_elt_t *task)
{

	log_debug("File:%s,Line:%d,Function:%s.task idx is %d..",__FILE__,__LINE__,__FUNCTION__,task->index);
	if (NULL == task)
	{
		log_error("File:%s,Line:%d,Function:%s.the thread argument is null.",
				__FILE__, __LINE__, __FUNCTION__);
		return;
	}

	if (0 >= task->sock)
	{
		log_error(
				"File:%s,Line:%d,Function:%s.the client socket fd is less 0.the task index is %d, sockfd is %d.",
				__FILE__, __LINE__, __FUNCTION__,task->index, task->sock);
		return;
	}

	if (-1 == set_nonblocking(task->sock))
	{
		log_error("File:%s,Line:%d,Function:%s.set the socket is error.",__FILE__,__LINE__,__FUNCTION__);
		return;
	}

	do
	{
		task->session_type = header;
		event_set(&(task->req_handler), task->sock, EV_READ, server_request_event,
				(void *) task);

		if (-1 == event_base_set(task->event_base, &(task->req_handler)))
		{
			log_error(
					"File:%s,Line:%d,Function:%s.Associate the read_header event to event_base is error.Errno:%d,Info:%s",
					__FILE__, __LINE__, __FUNCTION__,errno, strerror(errno));
			task->session_type = unspecified;
			break;
		}

		event_set(&(task->resp_handler), task->sock, EV_WRITE, server_response_event,
				(void *) task);

		if (0 != event_base_set(task->event_base, &(task->resp_handler)))
		{
			log_error(
					"File:%s,Line:%d,Function:%s.Associate the write hander to event_base is error.Errno:%d,Info:%s.",
					__FILE__, __LINE__, __FUNCTION__,errno,strerror(errno));
			task->session_type = unspecified;
			break;
		}

		if (0 != event_add(&(task->req_handler), &(task->timeout)))
		{
			log_error(
					"File:%s,Line:%d,Function:%s.Add the read header event to  event_base is error.Errno:%d,Info:%s.",
					__FILE__, __LINE__, __FUNCTION__, errno,strerror(errno));
			log_error("sockfd:%d.index:%d.",task->sock,task->index);
			task->session_type = unspecified;
			break;
		}

		event_base_loop(task->event_base, 0);
	} while (false);

	return;
}
